package com.hch.netty.mq;


import com.hch.netty.mq.entity.Message;
import com.hch.netty.mq.entity.User;
import com.hch.utils.JsonUtils;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.EventExecutorGroup;
import org.apache.commons.lang3.StringUtils;
import org.json.simple.JSONObject;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author heconghui
 */
public class NettyMQServer {

    private final int port;


    public NettyMQServer(int port) {
        this.port = port;
    }

    private void start() throws InterruptedException {
        // 1.bossGroup 线程用于接收连接，workerGroup 线程用于具体的处理
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            // 2.创建服务端启动引导/辅助类：ServerBootstrap
            ServerBootstrap b = new ServerBootstrap();
            // 3.给引导类配置两大线程组,确定了线程模型
            b.group(bossGroup, workerGroup)
                    // (非必备)打印日志
                    .handler(new LoggingHandler(LogLevel.INFO))
                    // 4.指定 IO 模型为 NIO
                    .channel(NioServerSocketChannel.class)
//            服务器端监听数据回调Handler
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) {
                            ChannelPipeline p = ch.pipeline();
                            // 5.可以自定义客户端消息的业务处理逻辑
                            p.addLast(new ChildrenHandler());
                        }
                    });
            // 6.绑定端口,调用 sync 方法阻塞直到绑定完成
            ChannelFuture f = b.bind(port).sync();
            System.out.println("网络服务端启动成功");
            //等待服务端监听端口关闭
            // 7.阻塞等待直到服务器Channel关闭 (closeFuture()方法获取Channel 的CloseFuture对象,然后调用sync()方法)
            f.channel().closeFuture().sync();
        } finally {
            // 8.优雅关闭相关线程组资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }


    /**
     *
     */
    private  BlockingQueue<String> message = new LinkedBlockingQueue<>();
    /**
     *
     */
    private static ArrayList<ChannelHandlerContext> consumerChannels = new ArrayList();

    private static final String type_consumer = "consumer";

    private static final String type_producer = "producer";


    private class ChildrenHandler extends ChannelInboundHandlerAdapter {
        // 服务端接收客户端发送的数据并响应
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            try {
                String body = byteBufToString(msg);
                Message<User> message = JsonUtils.fromJson(body, Message.class);
                String type = message.getType();
                switch (type) {
                    case type_consumer:
//                        消费者缓存链接通道，并主动获取消息队列
                        consumer(ctx);
                        return;
                    case type_producer:

                        String data = JsonUtils.toJson(message.getData());
//                        向消费者发送消息
                        producer(data);
                        return;
                }
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            } finally {
                ReferenceCountUtil.release(msg);

            }

        }

        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }

        // 处理客户端消息发生异常
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            // Close the connection when an exception is raised.
            System.out.println("异常处理");
            consumerChannels.remove(ctx);
            cause.printStackTrace();
            ctx.close();
        }
    }

    /**
     * 缓存生产者投递消息
     *
     * @param msg
     */
    private void producer(String msg) {
        // 缓存消息
        message.offer(msg);
        consumerChannels.forEach((ctx) -> {
            String poll = message.poll();
            if (StringUtils.isEmpty(poll)) {
                return;
            }
            // 发送数据给消费者
            ByteBuf resp = Unpooled.copiedBuffer(poll.getBytes());
            ctx.writeAndFlush(resp);
        });
        // 分组消息

    }

    /**
     * 消费者缓存链接通道，并主动获取消息队列
     *
     * @param ctx 通道
     */
    private void consumer(ChannelHandlerContext ctx) {
        // 保存连接
        consumerChannels.add(ctx);
        String poll = message.poll();
        if (!StringUtils.isEmpty(poll)) {
            // 从MQ服务器端获取消息给消费者
            ByteBuf resp = Unpooled.copiedBuffer(poll.getBytes());
            ctx.writeAndFlush(resp);
        }
        // 思路

    }

    private String byteBufToString(Object msg) throws UnsupportedEncodingException {
        if (msg == null) {
            return null;
        }
        ByteBuf buf = (ByteBuf) msg;
        byte[] req = new byte[buf.readableBytes()];
        buf.readBytes(req);
        String body = new String(req, "UTF-8");
        return body;
    }

    public static void main(String[] args) throws InterruptedException {
        new NettyMQServer(9008).start();

    }

}
